// @Author EthanScriptOn
// @Desc
package actuator

import (
	"fmt"
	"gitee.com/fatzeng/srf_switch_basic_components/common/custom_exp"
	"gitee.com/fatzeng/srf_switch_basic_components/common/proxy"
	"gitee.com/fatzeng/srf_switch_stream_bridge/bridge"
	"gitee.com/fatzeng/srf_switch_stream_bridge/framework/holder"
	"gitee.com/fatzeng/srf_switch_stream_bridge/framework/operator"
	"reflect"
	"sync"
)

type BridgeExecutor struct {
	bridgeExecutorProxy *proxy.Proxy
}

func GenerateBridgeExecutor() *BridgeExecutor {
	instance := new(BridgeExecutor)
	instance.bridgeExecutorProxy = proxy.ProxyTarget(instance)
	return instance
}

func (b *BridgeExecutor) GetProxy() *proxy.Proxy {
	return b.bridgeExecutorProxy
}

func (b *BridgeExecutor) Execute(ctx *SrfSwitchActuatorContext) (err error) {
	if ctx == nil {
		return fmt.Errorf("SrfSwitchActuatorContext is nil")
	}
	config := ctx.GetGroupConfig().BridgeConfigs
	if len(config) <= 0 {
		return fmt.Errorf("unable to find the list of prepared executions")
	}
	fence := sync.WaitGroup{}
	if !ctx.GetAsync() {
		fence.Add(len(config))
	}
	multiErr := new(custom_exp.MultiError)
	for _, bridgeConfigTemp := range config {
		bridgeConfig := bridgeConfigTemp
		go func(config *bridge.BridgeConfig) {
			defer b.fenceReduce(&fence, ctx.GetAsync())
			bridgeCtx := b.GenerateBridgeExecutorContext(config, holder.GenerateRecordHolder(ctx.stateTimeUnit))
			err = b.ExecuteSingle(bridgeCtx)
			if err != nil {
				multiErr.Errors = append(multiErr.Errors, err)
			}
		}(bridgeConfig)
	}
	fence.Wait()
	return multiErr
}

func (b *BridgeExecutor) GenerateBridgeExecutorContext(bridgeConfig *bridge.BridgeConfig, recordHolder *holder.RecordHolder) *SrfSwitchBridgeActuatorContext {
	return GenerateSrfSwitchBridgeActuatorContext(bridgeConfig, recordHolder)
}

func (b *BridgeExecutor) fenceReduce(fence *sync.WaitGroup, isAsync bool) {
	if !isAsync {
		fence.Done()
	}
}

func (b *BridgeExecutor) Record(holder *holder.RecordHolder, err *error) {
	holder.Record(err)
}

func (b *BridgeExecutor) ExecuteSingle(ctx *SrfSwitchBridgeActuatorContext) (err error) {
	config := ctx.BridgeConfig
	recordHolder := ctx.GetRecordHolder()
	defer b.Record(recordHolder, &err)
	structureParam, err := b.StructureParam(ctx, config)
	if err != nil {
		return err
	}
	operatorExecuteCall := b.GetProxy().Call(BridgeExecutor.OperatorExecute, ctx, config, structureParam)
	if err = operatorExecuteCall.GetError(); err != nil {
		return err
	}
	executeRes, _ := operatorExecuteCall.GetResultByType(interface{}(nil))
	resultParseCall := b.GetProxy().Call(BridgeExecutor.ResultParse, ctx, config, executeRes)
	if err = resultParseCall.GetError(); err != nil {
		return err
	}
	parserResult, _ := resultParseCall.GetResultByType(interface{}(nil))
	return b.GetProxy().Call(BridgeExecutor.BridgeCollectResult, ctx, config, parserResult).GetError()
}

func (b *BridgeExecutor) OperatorExecute(ctx *SrfSwitchBridgeActuatorContext, bridgeConfig *bridge.BridgeConfig, structureParam interface{}) (executeRes interface{}, err error) {
	operatorInstance := operator.GetOperator(bridgeConfig.MultiChannelDataOperatorIdentifier)
	if operatorInstance == nil {
		_ = ctx.Record(OperatorExecute, fmt.Sprintf("Unable to obtain operatorInstance with:%v", bridgeConfig.MultiChannelDataOperatorIdentifier))
		return nil, custom_exp.GenerateInterruptError(fmt.Sprintf("Unable to obtain operatorInstance with:%v", bridgeConfig.MultiChannelDataOperatorIdentifier))
	}
	executeRes, err = operatorInstance.Execute(ctx, bridgeConfig.OperatorConfigIdentifier, structureParam)
	if err != nil {
		_ = ctx.Record(OperatorExecute, fmt.Sprintf("extend_operator.Execute error: %v", err))
		return nil, custom_exp.GenerateFailError(fmt.Sprintf("extend_operator.Execute error: %v", err))
	}
	return
}

func (b *BridgeExecutor) ResultParse(ctx *SrfSwitchBridgeActuatorContext, bridgeConfig *bridge.BridgeConfig, executeRes interface{}) (parserResult interface{}, err error) {
	parserResult, err = bridgeConfig.ResultParser.Parse(ctx, bridgeConfig, executeRes)
	if err != nil {
		_ = ctx.Record(ResultAnalysis, fmt.Sprintf("ResultParser error: %v", err))
		return nil, custom_exp.GenerateFailError(fmt.Sprintf("config.ResultParser.Parse error: %v", err))
	}
	return
}

func (b *BridgeExecutor) BridgeCollectResult(ctx *SrfSwitchBridgeActuatorContext, config *bridge.BridgeConfig, parserResult interface{}) (err error) {
	err = config.ResultCollector.CollectResult(ctx, config, parserResult)
	if err != nil {
		_ = ctx.Record(ResultCollection, fmt.Sprintf("ResultCollector error: %v", err))
		return custom_exp.GenerateFailError(fmt.Sprintf("an exception occurred during the result collection process: %v", err))
	}
	return
}

func (b *BridgeExecutor) StructureParam(ctx *SrfSwitchBridgeActuatorContext, config *bridge.BridgeConfig) (structureParam interface{}, err error) {
	emptyParam := reflect.New(config.ParamType).Interface()
	err = b.GetProxy().Call(BridgeExecutor.ParamCollector, ctx, config, emptyParam).GetError()
	if err != nil {
		_ = ctx.Record(ParameterCollection, fmt.Sprintf("ParamCollector error: %v", err))
		return nil, custom_exp.GenerateInterruptError(fmt.Sprintf("an exception occurred during the parameter collection process: %v", err))
	}
	buildResult, err := config.ParamsBuilder.Build(emptyParam)
	if err != nil {
		_ = ctx.Record(ParameterConstruction, fmt.Sprintf("ParamsBuilder error: %v", err))
		return nil, custom_exp.GenerateInterruptError(fmt.Sprintf("an exception occurred during the parameter construction process: %v", err))
	}
	return buildResult, err
}

func (b *BridgeExecutor) ParamCollector(ctx *SrfSwitchBridgeActuatorContext, config *bridge.BridgeConfig, emptyParam interface{}) (err error) {
	collectors := config.ParamsCollectors()
	if len(collectors) <= 0 {
		_ = ctx.Record(ParameterCollection, fmt.Sprintf("parameterGatherer is nil"))
		return
	}
	for _, collector := range collectors {
		execute := collector.ShouldExecute()
		if !execute {
			continue
		}
		err = collector.Collect(emptyParam)
		if err != nil {
			return err
		}
	}
	return
}
